package cn.org.hentai.jtt1078.publisher;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.json.JSONUtil;
import cn.org.hentai.jtt1078.BO.VehicleStateBO;
import cn.org.hentai.jtt1078.annotation.FieldMapping;
import cn.org.hentai.jtt1078.codec.AudioCodec;
import cn.org.hentai.jtt1078.entity.AgricultureVideoLog;
import cn.org.hentai.jtt1078.entity.Media;
import cn.org.hentai.jtt1078.entity.MediaEncoding;
import cn.org.hentai.jtt1078.flv.FlvEncoder;
import cn.org.hentai.jtt1078.mapper.AgricultureVideoLogMapper;
import cn.org.hentai.jtt1078.subscriber.*;
import cn.org.hentai.jtt1078.util.ByteHolder;
import cn.org.hentai.jtt1078.util.Configs;
import cn.org.hentai.jtt1078.util.FLVUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by matrixy on 2020/1/11.
 */
public class Channel
{
    static Logger logger = LoggerFactory.getLogger(Channel.class);

    ConcurrentLinkedQueue<Subscriber> subscribers;
    RTMPPublisher rtmpPublisher;

    String tag;
    boolean publishing;
    ByteHolder buffer;
    AudioCodec audioCodec;
    FlvEncoder flvEncoder;
    private long firstTimestamp = -1;
    private long lastVideoFrameTimeOffset = 0;
    private boolean videoHeaderSent = false;
    private long videoTimestamp = 0;
    //第一次连接时的时间记录为视频文件对应时间
    private long timestamp = 0 ;
    //生成的视频文件所在路径
    private final String videoFilePath = "D:/iotx/Downloads/files/video/";
    //生成的视频文件后缀
    private final String suffix = ".flv";
    public final String vehicleStateFilePath = "D:\\17gps\\cache\\";

    public Channel(String tag)
    {
        this.tag = tag;
        this.subscribers = new ConcurrentLinkedQueue<Subscriber>();
        this.flvEncoder = new FlvEncoder(true, true);
        this.buffer = new ByteHolder(2048 * 1000);

        if (StringUtils.isEmpty(Configs.get("rtmp.url")) == false)
        {
            rtmpPublisher = new RTMPPublisher(tag);
            rtmpPublisher.start();
        }
    }

    public boolean isPublishing()
    {
        return publishing;
    }

    public Subscriber subscribe(ChannelHandlerContext ctx)
    {
        logger.info("channel: {} -> {}, subscriber: {}", Long.toHexString(hashCode() & 0xffffffffL), tag, ctx.channel().remoteAddress().toString());

        Subscriber subscriber = new VideoSubscriber(this.tag, ctx);
        this.subscribers.add(subscriber);
        return subscriber;
    }

    public void writeAudio(long timestamp, int pt, byte[] data)
    {
        if (audioCodec == null)
        {
            audioCodec = AudioCodec.getCodec(pt);
            logger.info("audio codec: {}", MediaEncoding.getEncoding(Media.Type.Audio, pt));
        }
        broadcastAudio(timestamp, audioCodec.toPCM(data));
    }

    public void writeVideo(SqlSession session,long sequence, long timeoffset, int payloadType, byte[] h264)
    {
        if (firstTimestamp == -1) firstTimestamp = timeoffset;
        this.publishing = true;
        this.buffer.write(h264);
        while (true)
        {
            byte[] nalu = readNalu();
            if (nalu == null) break;
            if (nalu.length < 4) continue;

            byte[] flvTag = this.flvEncoder.write(nalu, (int) (timeoffset - firstTimestamp));

            if (flvTag == null) continue;

            // 广播给所有的观众
            broadcastVideo(timeoffset, flvTag);

            //生成视频文件
            if (genVideoFile(session,timeoffset, flvTag)) return;
        }
    }


    private boolean genVideoFile(SqlSession session,long timeoffset, byte[] flvTag) {
        String bucket  = "video";
        String todayStr = DateUtil.format(new Date(), "yyyyMMdd");
        if (!new File(videoFilePath+todayStr).exists()) {
            new File(videoFilePath+todayStr).mkdirs();
        }
        if (timestamp == 0) {
            timestamp = timeoffset;
            // 获取映射器
            AgricultureVideoLogMapper mapper = session.getMapper(AgricultureVideoLogMapper.class);
            // 执行数据库操作
            AgricultureVideoLog agricultureVideoLog = new AgricultureVideoLog();
            String imei = tag.split("-")[0];
            agricultureVideoLog.setSimCode(imei);
            agricultureVideoLog.setChannelNo(Integer.parseInt(tag.split("-")[1]));
            agricultureVideoLog.setCreatedTime(new Date(timestamp));
            agricultureVideoLog.setUpdatedTime(new Date(timeoffset));
            agricultureVideoLog.setBucket(bucket);
            agricultureVideoLog.setDateStr(todayStr);
            agricultureVideoLog.setVideoFilePath(tag+"-"+timestamp+suffix);
            agricultureVideoLog.setWriteFlag(1);

            List<File> files = FileUtil.loopFiles(vehicleStateFilePath, (file) -> file.getName().startsWith("vhcTable_") && file.getName().endsWith(".cache"));
            files.forEach(file -> {
                FileUtil.readLines(file, StandardCharsets.UTF_8).forEach(line -> {
                        if (StringUtils.isNotBlank(line)) {
                            VehicleStateBO vehicleData = new VehicleStateBO();
                            try {
                                String[] pairs = line.split("#");
                                Map<String, String> map = new HashMap<>((int) ((float) pairs.length * 2 + 1.0F));
                                ObjectMapper objectMapper = new ObjectMapper();
                                for (String pair : pairs) {
                                    // 使用=分隔键和值
                                    String[] keyValue = pair.split("=", 2);
                                    String key = keyValue[0];
                                    String value = (keyValue.length > 1) ? keyValue[1] : "";
                                    if (StringUtils.isNotEmpty(value) && JSONUtil.isJsonObj(value)) {// JSON数据
                                        // JSON字符串转Map
                                        try {
                                            Map<String, String> valueMap = objectMapper.readValue(value, Map.class);
                                            if(!valueMap.isEmpty()){
                                                map.putAll(valueMap);
                                            }
                                        } catch (IOException e) {
                                            e.printStackTrace();
                                        }
                                    }
                                    map.put(key,value);
                                }
                                for (Field field : VehicleStateBO.class.getDeclaredFields()) {
                                    field.setAccessible(true);
                                    String key = field.isAnnotationPresent(FieldMapping.class) ?
                                            field.getAnnotation(FieldMapping.class).value() :
                                            field.getName();

                                    if (map.containsKey(key)) {
                                        try {
                                            field.set(vehicleData, convertToFieldType(field.getType(), MapUtils.getString(map,key)));
                                        } catch (IllegalAccessException e) {
                                            logger.error("设置字段值时访问冲突::", e);
                                        }
                                    }
                                }
                            } catch (Exception e) {
                                logger.error("解析车辆实时状态数据异常:", e);
                            }
                            Optional.ofNullable(vehicleData)
                                    .filter(data -> data.getImeiId().equals(imei))
                                    .ifPresent(data -> {
                                        // 如果vehicleData不为null且imeiId等于imei，则执行这里的代码
                                        agricultureVideoLog.setLat(vehicleData.getLat());
                                        agricultureVideoLog.setLng(vehicleData.getLng());
                                    });
                        }

                });
            });
            //vhcid保存
            Integer vhcid = mapper.getVhcidByImei(imei);
            if (vhcid!=null){
                agricultureVideoLog.setVhcid(vhcid+"");
            }
            mapper.insert(agricultureVideoLog);
        }
        String filePath = videoFilePath+todayStr+ File.separator+tag+"-"+timestamp+suffix;
        try (FileOutputStream fos = new FileOutputStream(filePath, true)) {
            // 之前是不是已经发送过了？没有的话，需要补发FLV HEADER的。。。
            if (lastVideoFrameTimeOffset == 0) lastVideoFrameTimeOffset = timeoffset;

            // 之前是不是已经发送过了？没有的话，需要补发FLV HEADER的。。。
            if (videoHeaderSent == false && flvEncoder.videoReady())
            {
                fos.write(flvEncoder.getHeader().getBytes());
                fos.write(flvEncoder.getVideoHeader().getBytes());

                // 直接下发第一个I帧
                byte[] iFrame = flvEncoder.getLastIFrame();
                if (iFrame != null)
                {
                    FLVUtils.resetTimestamp(iFrame, (int) videoTimestamp);
                    fos.write(iFrame);
                }

                videoHeaderSent = true;
            }

            if (flvTag == null) return true;

            // 修改时间戳
            FLVUtils.resetTimestamp(flvTag, (int) videoTimestamp);
            videoTimestamp += (int)(timeoffset - lastVideoFrameTimeOffset);
            lastVideoFrameTimeOffset = timeoffset;
            fos.write(flvTag);
        } catch (IOException e) {
            // 异常处理逻辑
            logger.info("Error writing to file: " + e.getMessage());
        }
        return false;
    }


    private static Object convertToFieldType(Class<?> fieldType, String value) {
        try {
            if (fieldType.equals(int.class) || fieldType.equals(Integer.class)) {
                return StringUtils.isNotEmpty(value)?Integer.parseInt(value):null;
            } else if (fieldType.equals(long.class) || fieldType.equals(Long.class)) {
                return StringUtils.isNotEmpty(value)?Long.parseLong(value):null;
            } else if (fieldType.equals(double.class) || fieldType.equals(Double.class)) {
                return StringUtils.isNotEmpty(value)?Double.parseDouble(value):null;
            } else if (fieldType.equals(LocalDateTime.class)) {
                try {
                    return StringUtils.isNotEmpty(value)?
                            LocalDateTime.parse(value, DateTimeFormatter.ofPattern("yyyy/M/d H:mm:ss")):null;
                } catch (Exception ignored) {
                }
            }
        } catch (NumberFormatException e) {
            logger.error("无法将字符串转换为数字: " + value + " 对于字段类型: " + fieldType.getName(), e);
        } catch (DateTimeParseException e) {
            logger.error("无法将字符串转换为日期: " + value + " 对于字段类型: " + fieldType.getName(), e);
        }
        return value;
    }

    public void broadcastVideo(long timeoffset, byte[] flvTag)
    {
        for (Subscriber subscriber : subscribers)
        {
            subscriber.onVideoData(timeoffset, flvTag, flvEncoder);
        }
    }

    public void broadcastAudio(long timeoffset, byte[] flvTag)
    {
        for (Subscriber subscriber : subscribers)
        {
            subscriber.onAudioData(timeoffset, flvTag, flvEncoder);
        }
    }

    public void unsubscribe(long watcherId)
    {
        for (Iterator<Subscriber> itr = subscribers.iterator(); itr.hasNext(); )
        {
            Subscriber subscriber = itr.next();
            if (subscriber.getId() == watcherId)
            {
                itr.remove();
                subscriber.close();
                return;
            }
        }
    }

    public void close()
    {
        for (Iterator<Subscriber> itr = subscribers.iterator(); itr.hasNext(); )
        {
            Subscriber subscriber = itr.next();
            subscriber.close();
            itr.remove();
        }
        if (rtmpPublisher != null) rtmpPublisher.close();
    }

    private byte[] readNalu()
    {
        for (int i = 0; i < buffer.size() - 3; i++)
        {
            int a = buffer.get(i + 0) & 0xff;
            int b = buffer.get(i + 1) & 0xff;
            int c = buffer.get(i + 2) & 0xff;
            int d = buffer.get(i + 3) & 0xff;
            if (a == 0x00 && b == 0x00 && c == 0x00 && d == 0x01)
            {
                if (i == 0) continue;
                byte[] nalu = new byte[i];
                buffer.sliceInto(nalu, i);
                return nalu;
            }
        }
        return null;
    }
}
